package com.example.repurchase.prediction.metrics;

import com.example.repurchase.prediction.UserRepurchasePrediction;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.time.Time;

public class RepurchaseMetrics {
    // 计算高价值用户数量（平均购买金额超过阈值）
    public static DataStream<Tuple2<Long, Integer>> calculateHighValueUsers(DataStream<UserRepurchasePrediction.UserRepurchaseScore> scores, double threshold) {
        return scores
                .filter(score -> score.getAvgPurchaseAmount() > threshold)
                .timeWindowAll(Time.minutes(1))
                .aggregate(new CountAggregate<>());
    }

    // 计算高复购可能性用户数量（复购分数超过阈值）
    public static DataStream<Tuple2<Long, Integer>> calculateHighProbabilityUsers(DataStream<UserRepurchasePrediction.UserRepurchaseScore> scores, double threshold) {
        return scores
                .filter(score -> score.getRepurchaseScore() > threshold)
                .timeWindowAll(Time.minutes(1))
                .aggregate(new CountAggregate<>());
    }

    // 计算平均复购分数
    public static DataStream<Tuple2<Long, Double>> calculateAverageRepurchaseScore(DataStream<UserRepurchasePrediction.UserRepurchaseScore> scores) {
        return scores
                .map((MapFunction<UserRepurchasePrediction.UserRepurchaseScore, Double>) UserRepurchasePrediction.UserRepurchaseScore::getRepurchaseScore)
                .timeWindowAll(Time.minutes(1))
                .aggregate(new AverageAggregate());
    }

    // 计数聚合函数
    public static class CountAggregate<T> implements AggregateFunction<T, Integer, Tuple2<Long, Integer>> {
        @Override
        public Integer createAccumulator() {
            return 0;
        }

        @Override
        public Integer add(T value, Integer accumulator) {
            return accumulator + 1;
        }

        @Override
        public Tuple2<Long, Integer> getResult(Integer accumulator) {
            return Tuple2.of(System.currentTimeMillis(), accumulator);
        }

        @Override
        public Integer merge(Integer a, Integer b) {
            return a + b;
        }
    }

    // 平均值聚合函数
    public static class AverageAggregate implements AggregateFunction<Double, Tuple2<Double, Integer>, Tuple2<Long, Double>> {
        @Override
        public Tuple2<Double, Integer> createAccumulator() {
            return Tuple2.of(0.0, 0);
        }

        @Override
        public Tuple2<Double, Integer> add(Double value, Tuple2<Double, Integer> accumulator) {
            return Tuple2.of(accumulator.f0 + value, accumulator.f1 + 1);
        }

        @Override
        public Tuple2<Long, Double> getResult(Tuple2<Double, Integer> accumulator) {
            return Tuple2.of(
                    System.currentTimeMillis(),
                    accumulator.f1 == 0 ? 0.0 : accumulator.f0 / accumulator.f1
            );
        }

        @Override
        public Tuple2<Double, Integer> merge(Tuple2<Double, Integer> a, Tuple2<Double, Integer> b) {
            return Tuple2.of(a.f0 + b.f0, a.f1 + b.f1);
        }
    }
}    